package com.bw.gmall.realtime.app.dws;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.app.func.DimAsyncFunction;
import com.bw.gmall.realtime.bean.TradeUserSpuOrderBean;
import com.bw.gmall.realtime.utils.*;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.*;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.sql.DriverManager;
import java.time.Duration;
import java.util.HashSet;
import java.util.List;
import java.util.concurrent.TimeUnit;

/*
*
下单各粒度：
   spu_id
   spu_name
   tm_id
   tm_name
   a3_id
   a3_name
   a2_id
   a2_name
   a1_id
   a1_name
   user_id
   orderCount
   amountCount
   *
   *
   1. 到  下单明细主题中加载数据
   2. 保留第一条数据减少数据量
   3. 把数据转换到对象里面（    spu_id   tm_id   a3_id）
   4. 关联habse  sku_info
   5. 分组聚合
   6. 关联hbase  spu
   6. 关联hbase  tm
   6. 关联hbase  a3
   6. 关联hbase  a2
   6. 关联hbase  a1

   1.旁路缓存
   2.异步IO
   3.多线程
   4.线程池
*
*
*
* */
public class DwsTradeUserSpuOrderWindow_b {

    public static void main(String[] args) throws Exception {

        //1.获取流失环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //2.设置并行度
        env.setParallelism(1);
        //3.设置检查点
        //4.加载数据
        DataStreamSource<String> addSource = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer("dwd_trade_order_detail", "DwsTradeUserSpuOrderWindow_b_xx"));

        //5.转换成json
        SingleOutputStreamOperator<JSONObject> map = addSource.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {

                JSONObject jsonObject = JSONObject.parseObject(value);


                out.collect(jsonObject);

            }
        });

        //6.去重
        SingleOutputStreamOperator<JSONObject> mapKeyBy = map.keyBy(a -> a.getString("id"))
                .process(new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                    ValueState<String> valueState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<String> valueStateDescriptor = new ValueStateDescriptor<>("valueState", String.class);
                        valueStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.seconds(5))
                                .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite).build());
                        valueState = getRuntimeContext().getState(valueStateDescriptor);
                    }

                    @Override
                    public void processElement(JSONObject value, Context ctx, Collector<JSONObject> out) throws Exception {

                        String value1 = valueState.value();
                        if (value1 == null) {
                            out.collect(value);
                            valueState.update("1");
                        }

                    }
                });

        SingleOutputStreamOperator<JSONObject> wt = mapKeyBy.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(2l))
                .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                    @Override
                    public long extractTimestamp(JSONObject element, long recordTimestamp) {
                        return DateFormatUtil.toTs(element.getString("create_time"), true);
                    }
                }));
        //7.通过sku_info  到hbase  关联  获取spu_id 和  tm_id  和 a3   userId   spu_id  tm_id  a3
        SingleOutputStreamOperator<TradeUserSpuOrderBean> map1 = wt.map(new RichMapFunction<JSONObject, TradeUserSpuOrderBean>() {
            DruidDataSource druidDataSource = null;

            @Override
            public void open(Configuration parameters) throws Exception {

                druidDataSource = DruidDSUtil.createDataSource();
            }

            @Override
            public TradeUserSpuOrderBean map(JSONObject value) throws Exception {

                //关联维度数据
                DruidPooledConnection connection = druidDataSource.getConnection();
                String sku_id = value.getString("sku_id");
                List<JSONObject> list = JdbcUtil.queryList(connection,
                        "select  *   from    GMALL_REALTIME.DIM_SKU_INFO  where   id='" + sku_id + "'"
                        , JSONObject.class, true);
                String tmId = list.get(0).getString("tmId");
                String category3Id = list.get(0).getString("category3Id");
                String spuId = list.get(0).getString("spuId");


                HashSet<String> set = new HashSet<>();
                set.add(value.getString("order_id"));


                return TradeUserSpuOrderBean.builder()
                        .category3Id(category3Id)
                        .spuId(spuId)
                        .trademarkId(tmId)
                        .skuId(sku_id)
                        .orderIdSet(set)
                        .userId(value.getString("user_id"))
                        .orderAmount(value.getDouble("split_total_amount"))
                        .ts(DateFormatUtil.toTs(value.getString("create_time"),true))
                        .build();
            }
        });

        map1.print("1----------------->");
        KeyedStream<TradeUserSpuOrderBean, Tuple4<String, String, String, String>> spuOrderBeanTuple4KeyedStream = map1.keyBy(new KeySelector<TradeUserSpuOrderBean, Tuple4<String, String, String, String>>() {
            @Override
            public Tuple4<String, String, String, String> getKey(TradeUserSpuOrderBean value) throws Exception {
                return Tuple4.of(
                        value.getUserId(),
                        value.getSpuId(),
                        value.getTrademarkId(),
                        value.getCategory3Id());
            }
        });

        //开窗聚合
        SingleOutputStreamOperator<TradeUserSpuOrderBean> reduce =
                spuOrderBeanTuple4KeyedStream.window(
                        TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10l)))
                .reduce(new ReduceFunction<TradeUserSpuOrderBean>() {
                    @Override
                    public TradeUserSpuOrderBean reduce(TradeUserSpuOrderBean value1, TradeUserSpuOrderBean value2) throws Exception {
                        value1.getOrderIdSet().addAll(value2.getOrderIdSet());

                        value1.setOrderAmount(value1.getOrderAmount() + value2.getOrderAmount());

                        return value1;
                    }
                }, new ProcessWindowFunction<TradeUserSpuOrderBean, TradeUserSpuOrderBean, Tuple4<String, String, String, String>, TimeWindow>() {
                    @Override
                    public void process(Tuple4<String, String, String, String> stringStringStringStringTuple4, Context context, Iterable<TradeUserSpuOrderBean> input, Collector<TradeUserSpuOrderBean> out) throws Exception {
                        TradeUserSpuOrderBean userSpuOrderBean = input.iterator().next();

                        userSpuOrderBean.setTs(System.currentTimeMillis());
                        userSpuOrderBean.setOrderCount((long) userSpuOrderBean.getOrderIdSet().size());
                        userSpuOrderBean.setStt(DateFormatUtil.toYmdHms(context.window().getStart()));
                        userSpuOrderBean.setEdt(DateFormatUtil.toYmdHms(context.window().getEnd()));

                        out.collect(userSpuOrderBean);
                    }
                });


        SingleOutputStreamOperator<TradeUserSpuOrderBean> tmName = reduce.map(new RichMapFunction<TradeUserSpuOrderBean, TradeUserSpuOrderBean>() {
            DruidDataSource druidDataSource = null;

            @Override
            public void open(Configuration parameters) throws Exception {

                druidDataSource = DruidDSUtil.createDataSource();
            }

            @Override
            public TradeUserSpuOrderBean map(TradeUserSpuOrderBean value) throws Exception {

                //关联维度数据
                DruidPooledConnection connection = druidDataSource.getConnection();
                String tmId = value.getTrademarkId();
                List<JSONObject> list = JdbcUtil.queryList(connection,
                        "select  *   from    GMALL_REALTIME.DIM_BASE_TRADEMARK  where   id='" + tmId + "'"
                        , JSONObject.class, true);

                String tmName = list.get(0).getString("tmName");


                return TradeUserSpuOrderBean.builder()
                        .trademarkName(tmName)
                        .build();
            }
        });
        SingleOutputStreamOperator<TradeUserSpuOrderBean> spuName = tmName.map(new RichMapFunction<TradeUserSpuOrderBean, TradeUserSpuOrderBean>() {
            DruidDataSource druidDataSource = null;

            @Override
            public void open(Configuration parameters) throws Exception {

                druidDataSource = DruidDSUtil.createDataSource();
            }

            @Override
            public TradeUserSpuOrderBean map(TradeUserSpuOrderBean value) throws Exception {

                //关联维度数据
                DruidPooledConnection connection = druidDataSource.getConnection();
                String spuId = value.getSpuId();
                List<JSONObject> list = JdbcUtil.queryList(connection,
                        "select  *   from  GMALL_REALTIME.DIM_SPU_INFO   where   id='" + spuId + "'"
                        , JSONObject.class, true);

                String spuName = list.get(0).getString("spuName");


                return TradeUserSpuOrderBean.builder()
                        .spuName(spuName)
                        .build();
            }
        });

        SingleOutputStreamOperator<TradeUserSpuOrderBean> a3 = spuName.map(new RichMapFunction<TradeUserSpuOrderBean, TradeUserSpuOrderBean>() {
            DruidDataSource druidDataSource = null;

            @Override
            public void open(Configuration parameters) throws Exception {

                druidDataSource = DruidDSUtil.createDataSource();
            }

            @Override
            public TradeUserSpuOrderBean map(TradeUserSpuOrderBean value) throws Exception {

                //关联维度数据
                DruidPooledConnection connection = druidDataSource.getConnection();
                String spuId = value.getSpuId();
                List<JSONObject> list = JdbcUtil.queryList(connection,
                        "select  *   from  GMALL_REALTIME.DIM_BASE_CATEGORY3   where   id='" + spuId + "'"
                        , JSONObject.class, true);

                String name = list.get(0).getString("name");
                String category2Id = list.get(0).getString("category2Id");


                return TradeUserSpuOrderBean.builder()
                        .category2Id(category2Id)
                        .category3Name(name)
                        .build();
            }
        });
        a3.print("---->");
        env.execute();
        //keyby 分组统计   2024-04-23 18:59:26   2024-04-23 19:00:00














    }

}